package com.offcn.bigdata.spark.p3

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 广播变量操作
  *     一个非常重要的作用，就是讲reduce join转化为map的join，避免的shuffle操作，提升了程序性能。
  *     但是此种操作不适合广播大变量和频繁更新的变量；适合大小表关联的操作。
  *  reduce join就是说白了，调用join函数，有shuffle操作。
  *  map join就是对其中的一个rdd进行filter/map/flatMap等类map算子操作是，和另外一张表(不是rdd)完成关联
  *
  */
object _01BroadcastOps {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf()
            .setMaster("local[*]")
            .setAppName(s"${_01BroadcastOps.getClass.getSimpleName}")
        val sc = new SparkContext(conf)

        case class Student(id: Int, name: String, age: Int, gender: Int)
        case class Score(sid: Int, course: String, score: Double)
        val list = List(
            Student(1, "林博", 18, 0),
            Student(2, "单松", 19, 1),
            Student(3, "张皓", 20, 0),
            Student(4, "王建", 20, 1),
            Student(106, "冯岩", 30, 1)
        )

        val stus = sc.parallelize(list)
        val scores = sc.parallelize(List(
            Score(2, "语文", 80),
            Score(3, "数学", 70.5),
            Score(1, "英语", 30.5),
            Score(4, "体育", 60),
            Score(105, "信息技术", 100)
        ))

        // 关联学习信息及其考试成绩
        val id2Stu:RDD[(Int, Student)] = stus.map(stu => (stu.id, stu))
        val id2Score:RDD[(Int, Score)] = scores.map(score => (score.sid, score))
        println("------------join结果--------------")
        val joinedRDD:RDD[(Int, (Student, Score))] = id2Stu.join(id2Score)//shuffle操作
        joinedRDD.foreach{case (id, (stu, score)) => {
            println(s"学生id：${id}, name: ${stu.name}, 性别：${stu.gender}, 科目：${score.course}, 成绩: ${score.score}")
        }}

        println("------------broadcast结果--------------")// map join
        val id2Map: Map[Int, Student] = list.map(stu => (stu.id, stu)).toMap
        //构造广播变量
        val stuBC:Broadcast[Map[Int, Student]] = sc.broadcast(id2Map)

        scores.foreach(score => {
            //获取广播变量
            val bc = stuBC.value
            val stu = bc.getOrElse(score.sid, null)
            println(s"学生id：${score.sid}, 学生信息：${stu}, 科目：${score.course}, 成绩: ${score.score}")
        })

        sc.stop()
    }
}
